package com.atguigu.flink.day08;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/8/19
 * 该案例演示了定时器
 */
public class Flink02_Timer {
    public static void main(String[] args) throws Exception {
        //TODO 1.指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.设置并行度
        env.setParallelism(1);
        //TODO 3.从指定的网络端口读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 4.对流中数据进行类型转换    String->WaterSensor
        SingleOutputStreamOperator<WaterSensor> wsDS = socketDS.map(
               new WaterSensorMapFunction()
        );

        wsDS.process(
                new ProcessFunction<WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor ws, ProcessFunction<WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        TimerService timerService = ctx.timerService();
                        timerService.registerProcessingTimeTimer(timerService.currentProcessingTime() + 5000);
                    }

                    @Override
                    public void onTimer(long timestamp, ProcessFunction<WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        System.out.println("定时器被触发了");
                    }
                }
        ).print();

        /*//指定Watermark的生成策略，并提取事件时间字段
        SingleOutputStreamOperator<WaterSensor> withWatermarkDS = wsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<WaterSensor>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<WaterSensor>() {
                                    @Override
                                    public long extractTimestamp(WaterSensor ws, long recordTimestamp) {
                                        return ws.getTs();
                                    }
                                }
                        )
        );

        //TODO 5.按照传感器的id对流中数据进行分组
        KeyedStream<WaterSensor, String> keyedDS = withWatermarkDS.keyBy(WaterSensor::getId);
        //TODO 6.演示定时器的使用
        SingleOutputStreamOperator<String> processDS = keyedDS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor ws, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        //获取当前分组的key
                        String currentKey = ctx.getCurrentKey();
                        System.out.println("currentKey:" + currentKey);

                        //获取当前元素的事件时间
                        Long timestamp = ctx.timestamp();
                        System.out.println("timestamp:" + timestamp);

                        //注意：如果要使用侧输出流，必须用process算子，因为只有在process底层才能调用ctx.output
                        //ctx.output();

                        //获取定时服务
                        TimerService timerService = ctx.timerService();

                        //获取当前处理时间
                        long currentProcessingTime = timerService.currentProcessingTime();
                        System.out.println("currentProcessingTime:" + currentProcessingTime);

                        //获取当前水位线信息
                        long currentWatermark = timerService.currentWatermark();
                        System.out.println("currentWatermark:" + currentWatermark);

                        //注册处理时间定时器
                        timerService.registerProcessingTimeTimer(currentProcessingTime + 10000);
                        //注册事件时间定时器
                        //timerService.registerEventTimeTimer(10);

                        //删除处理时间定时器
                        //timerService.deleteProcessingTimeTimer();

                        //删除事件时间定时器
                        //timerService.deleteEventTimeTimer(10);

                    }

                    //当定时器被触发的时候执行该方法
                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        System.out.println("传感器id"+ctx.getCurrentKey()+"是在"+timestamp+"被触发了");
                    }
                }
        );*/
        //processDS.print();

        env.execute();

    }
}
